import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class MyThreadPool {

    private final int corePoolSize;     // 核心线程数
    private final int maximumPoolSize;  // 最大线程数
    private final long keepAliveTime;   // 线程空闲时间
    private BlockingQueue<Runnable> taskQueue; // 任务队列
    private ThreadFactory threadFactory;       // 线程工厂
    private RejectedExecutionHandler handler;  // 拒绝策略
    // 线程计数器和线程集合
    private final AtomicInteger threadCount = new AtomicInteger();
    private final Worker[] workers;
    // 线程池状态
    private volatile boolean isShutdown = false;

    // 构造函数
    public MyThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                        BlockingQueue<Runnable> taskQueue,
                        ThreadFactory threadFactory,
                        RejectedExecutionHandler handler) {
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.keepAliveTime = keepAliveTime;
        this.taskQueue = taskQueue;
        this.threadFactory = threadFactory;
        this.handler = handler;
        this.workers = new Worker[maximumPoolSize];
    }

    public void execute(Runnable task) {
        if (Objects.isNull(task)) throw new NullPointerException("task is null");
        if (isShutdown) {
            handler.rejectedExecution(task, this);
            return;
        }
        // 1. 如果当前线程数小于核心线程数，创建新线程执行任务
        int currentCount = threadCount.get();
        if (currentCount <= corePoolSize) {
            if (addWorker(task, true)) return;
            // 创建失败可能情况是因为线程池已经关闭，或者已到达线程池最大线程数
        }
        //2. 核心线程已满。将任务加入队列
        if (taskQueue.offer(task)) {
            //检擦是否又有空闲核心线程
            if (threadCount.get() == 0) {
                addWorker(null, false);
            }
        }
        //3.队列已满。尝试创建新线程
        else if (!addWorker(task, false)) {
            // 进来表示已经达到最大线程数了，执行拒绝策略
            handler.rejectedExecution(task, this);
        }
    }

    // 创建一个线程对象去执行任务
    private final class Worker implements Runnable {
        private Thread thread;
        private Runnable firstTask;

        public Worker(Runnable firstTask) {
            this.firstTask = firstTask;
        }

        @Override
        public void run() {
            Runnable task = firstTask;
            firstTask = null;
            try {
                //不断从任务队列中获取任务执行
                while (task != null || (task = getTask()) != null) {
                    task.run();
                    task = null;
                }
            } finally {
                //工作线程退出时，更新线程计数
                //获取外部类MyThreadPool当前实例的监视器锁
                //在同一时刻，只有一个线程能够进入被这个锁保护的代码块
                //要是多个线程同时尝试访问这个代码块，其他线程就会被阻塞，直到持有锁的线程执行完同步块中的代码并释放锁。
                synchronized (MyThreadPool.this) {
                    //线程集合操作,表示对应位置线程无空闲时间
                    for (int i = 0; i < workers.length; i++) {
                        if (workers[i] != this) {
                            workers[i] = null;
                            break;
                        }
                    }
                    //减少一个可用线程数
                    threadCount.decrementAndGet();
                }
            }
        }

        // 从队列中获取任务，支持超时机制
        private Runnable getTask() {
            try {
                return threadCount.get() > corePoolSize ? taskQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : taskQueue.take();
            } catch (InterruptedException e) {
                //打断正常运行的线程, 线程并不会暂停
                // 只是调用方法Thread.currentThread().isInterrupted();的返回值为true
                // 可以判断Thread.currentThread().isInterrupted();的值来手动停止线程
                Thread.currentThread().interrupt();
                return null;
            }
        }

        // 中断工作线程
        public void interruptIfStarted() {
            Thread t = thread;
            if (t != null && !t.isInterrupted()) {
                t.interrupt();
            }
        }
    }

    public boolean addWorker(Runnable task, boolean isCore) {
        if (isShutdown) return false;
        int currentCount = threadCount.get();
        // 校验线程数是否超出规定
        if ((isCore && currentCount >= corePoolSize) || (!isCore && currentCount >= maximumPoolSize)) {
            return false;
        }
        //创建工作线程
        Worker worker = new Worker(task);
        Thread t = threadFactory.newThread(worker);
        worker.thread = t;
        synchronized (this) {
            if (isShutdown) return false;
            //放入线程集合
            for (int i = 0; i < workers.length; i++) {
                if (workers[i] == null) {
                    workers[i] = worker;
                    break;
                }
            }
            threadCount.incrementAndGet();
        }
        t.start();
        return true;
    }

    // -------------------------------------- 线程工厂-------------------------------------
    // 线程工厂接口
    public interface ThreadFactory {
        Thread newThread(Runnable r);
    }

    // 默认线程工厂
    public static class DefaultThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        public DefaultThreadFactory() {
            group = Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-";
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
            if (t.isDaemon()) t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY) t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }

    // --------------------------------------拒绝策略--------------------------------------
    // 拒绝策略接口
    public interface RejectedExecutionHandler {
        void rejectedExecution(Runnable r, MyThreadPool executor);
    }

    // 丢弃任务并抛出异常的拒绝策略
    public static class AbortPolicy implements RejectedExecutionHandler {
        @Override
        public void rejectedExecution(Runnable r, MyThreadPool executor) {
            throw new java.util.concurrent.RejectedExecutionException("Task " + r.toString() + " rejected from " + executor.toString());
        }
    }

    // 关闭线程池
    public void shutdown() {
        synchronized (this) {
            isShutdown = true;
            // 中断所有工作线程
            for (Worker worker : workers) {
                if (worker != null) {
                    worker.interruptIfStarted();
                }
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        MyThreadPool pool = new MyThreadPool(
                2,                      // 核心线程数
                5,                      // 最大线程数
                60,                     // 空闲时间
                new LinkedBlockingQueue<>(10), // 任务队列
                new DefaultThreadFactory(),    // 线程工厂
                new AbortPolicy()              // 拒绝策略
        );

        // 提交任务
        for (int i = 0; i < 20; i++) {
            final int taskId = i;
            try {
                pool.execute(() -> {
                    System.out.println("Executing task " + taskId + " by " + Thread.currentThread().getName());
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                    System.out.println("Completed task " + taskId);
                });
            } catch (Exception e) {
                System.out.println("Task " + taskId + " rejected: " + e.getMessage());
            }
        }

        // 关闭线程池
        Thread.sleep(2000);
        pool.shutdown();
    }
}
